/**
 * Copyright 2009 Erick Armbrust 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0 
 *   
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */

package tapioca.dt;

import java.util.List;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.repackaged.com.google.common.base.Preconditions;
import com.google.appengine.repackaged.com.google.common.collect.Lists;

/**
 * A simple wrapper around an {@link Entity} that contains all the data for a distributed 
 * transaction.
 *
 * @author armbrust@gmail.com (Erick Armbrust)
 */
public class DistributedTransactionEntity {

  public static final String KIND = "DT__";
  public static final String CREATED_TIMESTAMP_PROP = "createdTimestamp";
  public static final String STATE_PROP = "state";
  public static final String GET_KEYS_PROP = "getKeys";
  public static final String GET_VERSIONS_PROP = "getVersions";
  public static final String PUT_KEYS_PROP = "putKeys";
  public static final String PUT_SHADOW_KEYS_PROP = "putShadowKeys";
  public static final String PUT_CREATION_FLAGS_PROP = "putCreationFlags";
  
  private final Entity entity;
  private final long createdTimestamp;
  private State state;
  private List<GetRequest> gets;
  private List<PutRequest> puts;
  
  /**
   * 
   */
  public DistributedTransactionEntity() {
    this.entity = new Entity(KIND);
    createdTimestamp = System.currentTimeMillis();
    state = State.INIT_0;
    gets = Lists.newArrayList();
    puts = Lists.newArrayList();
  }
  
  /**
   * 
   * @param entity
   */
  @SuppressWarnings("unchecked")
  public DistributedTransactionEntity(Entity entity) {
    Preconditions.checkArgument(entity.getKind().equals(KIND), 
        "Entity with kind " + entity.getKind() + " is not a valid DT entity.");
    
    this.entity = entity;
    createdTimestamp = (Long) entity.getProperty(CREATED_TIMESTAMP_PROP);
    state = State.getValue((String) entity.getProperty(STATE_PROP));

    gets = Lists.newArrayList();
    List<Key> getKeys = (List<Key>) entity.getProperty(GET_KEYS_PROP);
    List<Key> getVersions = (List<Key>) entity.getProperty(GET_VERSIONS_PROP);
    if (getKeys != null && getVersions != null) {
      if (getKeys.size() != getVersions.size()) {
        throw new RuntimeException("GetRequest list sizes do not match");
      } else {
        for (int i = 0; i < getKeys.size(); i++) {
          gets.add(new GetRequest(getKeys.get(i), getVersions.get(i)));
        }
      }
    }
    
    puts = Lists.newArrayList();
    List<Key> putKeys = (List<Key>) entity.getProperty(PUT_KEYS_PROP);
    List<Key> putShadowKeys = (List<Key>) entity.getProperty(PUT_SHADOW_KEYS_PROP);
    List<Boolean> putCreationFlags = (List<Boolean>) entity.getProperty(PUT_CREATION_FLAGS_PROP);
    if (putKeys != null && putShadowKeys != null) {
      if (putKeys.size() != putShadowKeys.size() && putKeys.size() != putCreationFlags.size()) {
        throw new RuntimeException("PutRequest list sizes do not match");
      } else {
        for (int i = 0; i < putKeys.size(); i++) {
          puts.add(new PutRequest(putKeys.get(i), putShadowKeys.get(i), putCreationFlags.get(i)));
        }
      }
    }
  }

  public Entity getEntity() {
    return entity;
  }

  public long getCreatedTimestamp() {
    return createdTimestamp;
  }
  
  public State getState() {
    return state;
  }

  public void setState(State state) {
    this.state = state;
  }

  public List<GetRequest> getGets() {
    return gets;
  }

  public void setGets(List<GetRequest> gets) {
    this.gets = gets;
  }

  public void addGet(GetRequest get) {
    gets.add(get);
  }
  
  public List<PutRequest> getPuts() {
    return puts;
  }

  public void setPuts(List<PutRequest> puts) {
    this.puts = puts;
  }
  
  public void addPut(PutRequest put) {
    puts.add(put);
  }
  
  /**
   * 
   */
  void fillEntity() {
    entity.setProperty(CREATED_TIMESTAMP_PROP, createdTimestamp);
    entity.setProperty(STATE_PROP, state.toString());

    List<Key> getKeys = Lists.newArrayList();
    List<Key> getVersions = Lists.newArrayList();
    for (GetRequest req : gets) {
      getKeys.add(req.getKey());
      getVersions.add(req.getVersion());
    }
    entity.setProperty(GET_KEYS_PROP, getKeys);
    entity.setProperty(GET_VERSIONS_PROP, getVersions);
    
    List<Key> putKeys = Lists.newArrayList();
    List<Key> putShadowKeys = Lists.newArrayList();
    List<Boolean> putCreationFlags = Lists.newArrayList();
    for (PutRequest putRequest : puts) {
      putKeys.add(putRequest.getKey());
      putShadowKeys.add(putRequest.getShadowKey());
      putCreationFlags.add(putRequest.isNonNamedCreateRequest());
    }
    entity.setProperty(PUT_KEYS_PROP, putKeys);
    entity.setProperty(PUT_SHADOW_KEYS_PROP, putShadowKeys);    
    entity.setProperty(PUT_CREATION_FLAGS_PROP, putCreationFlags);
  }
  
  /**
   * 
   * @param ds
   */
  public void save(DatastoreService ds) {
    save(ds, null);
  }
  
  /**
   * 
   * @param ds
   * @param txn
   */
  public void save(DatastoreService ds, Transaction txn) {
    fillEntity();
    
    if (txn == null) {
      ds.put(entity);
    } else {
      ds.put(txn, entity);
    }    
  }
  
}
